package com.jerry.service.Impl;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jerry.BO.CompleteChunkFileBO;
import com.jerry.BO.FileChunkBO;
import com.jerry.BO.FileChunkMeta;
import com.jerry.BO.MetaFile;
import com.jerry.DTO.FileMeta;
import com.jerry.VO.FileChunkMetaVO;
import com.jerry.VO.MetaFileVO;
import com.jerry.config.ClientConfig;
import com.jerry.errors.BusinessException;
import com.jerry.errors.EnumException;
import com.jerry.response.CommonResponse;
import com.jerry.service.FileService;
import com.jerry.utils.ChunkAddressStrategy;


import com.jerry.utils.Md5Util;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@Slf4j
@Service
public class FileServiceImpl implements FileService {
    private final ClientConfig clientConfig;
    private final RestTemplate restTemplate;
    private final ObjectMapper objectMapper;
    private final ChunkAddressStrategy chunkAddressStrategy;


    public FileServiceImpl(ClientConfig clientConfig, RestTemplate restTemplate, ObjectMapper objectMapper, ChunkAddressStrategy chunkAddressStrategy, MongoTemplate mongoTemplate) {
        this.clientConfig = clientConfig;
        this.restTemplate = restTemplate;
        this.objectMapper = objectMapper;
        this.chunkAddressStrategy = chunkAddressStrategy;

    }

    @Override
    public MetaFile getMeta(String bucketName, String filename) {
        if (filename.contains(".")) {
            filename = filename.split("\\.")[0];
        }
        String metaUrl = clientConfig.getMetaServerAddress() + "/meta/info?bucketName={bucketName}&filename={filename}";
        Map params = new HashMap<String, Object>();
        params.put("bucketName", bucketName);
        params.put("filename", filename);
        Object res = restTemplate.getForObject(metaUrl, Object.class, params);
        CommonResponse<MetaFile> commonResponse = objectMapper.convertValue(res, new TypeReference<CommonResponse<MetaFile>>() {
        });
        return commonResponse.getData();
    }

    @Override
    public String upload(String bucketName, MultipartFile file, String nickName, String fileType) throws IOException {
        String originalFilename = file.getOriginalFilename();
        log.info("{}", originalFilename);
        String extension = "";
        if (originalFilename != null) {
            int indexOf = originalFilename.lastIndexOf(".");
            if (indexOf != -1) {
                extension = originalFilename.substring(indexOf);
            }
        }
        /*FileMeta fileMeta = new FileMeta()
                .setBucketName(bucketName)
                .setExtension(extension)
                .setFileSize(file.getSize());
        log.info("{}",fileMeta);*/
        Map<String, Object> params = new HashMap<>();
        params.put("fileSize", file.getSize());
        params.put("extension", extension);
        params.put("bucketName", bucketName);
        params.put("nickName", nickName);
        Object response = restTemplate.postForObject(clientConfig.getMetaServerAddress() + "/meta/generate",
                params, Object.class);
        CommonResponse<MetaFile> metaFileCommonResponse = objectMapper
                .convertValue(response, new TypeReference<CommonResponse<MetaFile>>() {
                });
        if (metaFileCommonResponse == null) {
            throw new RuntimeException("获取meta失败");
        }
        MetaFile metaFile = metaFileCommonResponse.getData();
        if (metaFile == null) {
            throw new RuntimeException("meta为空");
        }
        try {
            uploadChunks(file, metaFile);
        } catch (Exception e) {
            e.printStackTrace();
        }
       /* try {
            uploadChunks(file, metaFile);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("上传分片失败");
        }*/
        return String.format("%s/%s.%s", metaFile.getBucketName(), metaFile.getFileName(), metaFile.getExtension());
    }

    @Scheduled(cron = "*/30 * * * * *")
    public void splice() {
        String url = clientConfig.getMetaServerAddress() + "/meta/files/video";
        Object response = restTemplate.getForObject(url, Object.class);
        CommonResponse<List<MetaFile>> commonResponse = objectMapper.convertValue(response,new TypeReference<CommonResponse<List<MetaFile>>>(){
        });
        List<MetaFile> metaFiles = commonResponse.getData();
        metaFiles = metaFiles.stream()
                .filter(metaFile -> {
                    for (FileChunkMeta fileChunkMeta : metaFile.getFileChunkMetas()) {
                        if (!fileChunkMeta.getCompleted())
                            return false;
                    }
                    return true;
                })
                .collect(Collectors.toList());
        for (MetaFile metaFile : metaFiles) {
            List<FileChunkMeta> chunks = metaFile.getFileChunkMetas();
            String fileName = metaFile.getFileName();
            String extension = metaFile.getExtension();
            File file = new File(clientConfig.getWorkSpace() + fileName + extension);
            if (!file.exists()) {
                if (!file.exists()) {
                    try {
                        file.createNewFile();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                } else {
                    log.info("File already exists.");
                    return;
                }
            }
            try {
                FileOutputStream fileOutputStream = new FileOutputStream(file);
                for (FileChunkMeta chunk : chunks) {
                    byte[] content = this.downloadChunk(chunk);
                    try {
                        fileOutputStream.write(content);
                    } catch (IOException e) {
                        e.printStackTrace();
                        throw new BusinessException(EnumException.FALE_TO_DOWNLOAD_FILE);
                    }
                }
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        }
        log.info("已拼接任务");
    }

    @Override
    public byte[] downloadChunk(FileChunkMeta chunk) {
        String url = chunkAddressStrategy.get(chunk);
        Map params = new HashMap<String, Object>();
        params.put("bucketName", chunk.getBucketName());
        params.put("filename", chunk.getFilename());
        params.put("extension", chunk.getExtension());
        params.put("chunkNo", chunk.getChunkNo());
        Object res = restTemplate.getForObject(url + "/file/read?bucketName={bucketName}" +
                "&filename={filename}&extension={extension}&chunkNo={chunkNo}", Object.class, params);
        CommonResponse<byte[]> commonResponse = objectMapper.convertValue(res, new TypeReference<CommonResponse<byte[]>>() {
        });
        return commonResponse.getData();
    }

    @Override
    public MetaFileVO  meta(FileMeta fileMeta) {
        String url = clientConfig.getMetaServerAddress() + "/meta/generate";
        Object response = restTemplate.postForObject(url, fileMeta, Object.class);
        if (response == null) {
            throw new BusinessException(EnumException.FALE_TO_GENERATE_METAFILE);
        }
        CommonResponse<MetaFile> commonResponse = objectMapper.convertValue(response, new TypeReference<CommonResponse<MetaFile>>() {
        });
        MetaFile metaFile = commonResponse.getData();
        return buildMetaFileVO(metaFile);
    }

    @Override
    public String uploadChunk(String bucketName,
                              String fileName,
                              String md5,
                              Integer chunkNo,
                              MultipartFile file) {
        /*MetaFile metaFile = getMeta(bucketName, filename);
        List<FileChunkMeta> chunks = metaFile.getFileChunkMetas();*/
        String metaServerAddress = clientConfig.getMetaServerAddress();
        String url = metaServerAddress + "/meta/chunk/info?bucketName={bucketName}&fileName={fileName}&chunkNo={chunkNo}";
        Map<String, Object> params = new HashMap<>();
        params.put("bucketName", bucketName);
        params.put("fileName", fileName);
        params.put("chunkNo", chunkNo);
        Object resp = restTemplate.getForObject(url, Object.class, params);
        CommonResponse<List<FileChunkMeta>> chunkInfoResp = objectMapper.convertValue(resp, new TypeReference<CommonResponse<List<FileChunkMeta>>>() {
        });
        List<FileChunkMeta> chunks = chunkInfoResp.getData();
        String realMd5 = Md5Util.getMd5(file);
        if (!realMd5.equals(md5)) {
            throw new BusinessException(EnumException.CHUNK_FILE_INCOMPLETE);
        }
        chunks = chunks.stream().filter(c -> c.getChunkNo().equals(chunkNo)).collect(Collectors.toList());
        chunks.forEach(c -> {
            int chunkSize = c.getChunkSize();
            byte[] buffer = new byte[chunkSize];
            try (InputStream inputStream = file.getInputStream()) {
                inputStream.read(buffer);
                FileChunkBO fileChunkBO = new FileChunkBO();
                fileChunkBO.setFilename(c.getFilename())
                        .setExtension(c.getExtension())
                        .setChunksize(c.getChunkSize())
                        .setChunkNo(c.getChunkNo())
                        .setBucketName(c.getBucketName())
                        .setBytes(buffer);
                String address = chunkAddressStrategy.get(c);
                Object response = restTemplate.postForObject(
                        address + "/file/write",
                        fileChunkBO,
                        Object.class
                );
                if (response == null) {
                    throw new RuntimeException(String.format("第%s分片上传失败", c.getChunkNo()));
                }
                CommonResponse<String> commonResponse = objectMapper.convertValue(response, new TypeReference<CommonResponse<String>>() {
                });
                String serverMd5 = commonResponse.getData();
                if (!serverMd5.equals(realMd5)) {
                    throw new RuntimeException(String.format("第%s分片不完整", c.getChunkNo()));
                }
                CompleteChunkFileBO completeChunkFileBO = new CompleteChunkFileBO();
                completeChunkFileBO.setAddress(c.getAddress())
                        .setChunkNo(c.getChunkNo())
                        .setFilename(c.getFilename())
                        .setSchema(c.getSchema())
                        .setMd5(md5);
                Object res = restTemplate.postForObject(metaServerAddress + "/meta/chunk/complete",
                        completeChunkFileBO,
                        Object.class
                );
                if (res == null) {
                    throw new RuntimeException("修改metafile失败");
                }
            } catch (Exception e) {
                throw new BusinessException(EnumException.FALE_TO_UPLOAD_FILE);
            }
        });
        return md5;
    }

    @Override
    public void delete(String bucketName, String fileName) {
        String metaServerAddress = clientConfig.getMetaServerAddress();
        String url = String.format("%s/meta/%s/%s", metaServerAddress, bucketName, fileName);
        restTemplate.delete(url);
        MetaFile metaFile = getMeta(bucketName, fileName);
        List<FileChunkMeta> fileChunkMetas = metaFile.getFileChunkMetas();
        for (FileChunkMeta fileChunkMeta : fileChunkMetas) {
            String extension = fileChunkMeta.getExtension();
            Integer chunkNo = fileChunkMeta.getChunkNo();
            String address = fileChunkMeta.getAddress();
            Map params = new HashMap();
            params.put("filename",fileName);
            params.put("extension",extension);
            params.put("bucketName",bucketName);
            params.put("chunkNo",chunkNo);
            restTemplate.delete(address + "/file/delete",params);
        }

    }

    private MetaFileVO buildMetaFileVO(MetaFile metaFile) {
        MetaFileVO metaFileVO = new MetaFileVO();
        List<FileChunkMeta> chunks = metaFile.getFileChunkMetas();
        List<FileChunkMetaVO> fileChunkMetaVOS = new ArrayList<>();
        for (FileChunkMeta chunk : chunks) {
            FileChunkMetaVO fileChunkMetaVO = new FileChunkMetaVO();
            fileChunkMetaVO.setChunkNo(chunk.getChunkNo())
                    .setChunkSize(chunk.getChunkSize())
                    .setChunkStart(chunk.getChunkStart())
                    .setFilename(chunk.getFilename())
                    .setCompleted(chunk.getCompleted());
            fileChunkMetaVOS.add(fileChunkMetaVO);
        }
        fileChunkMetaVOS = fileChunkMetaVOS.stream().distinct().collect(Collectors.toList());
        return metaFileVO.setBucketName(metaFile.getBucketName())
                .setFilename(metaFile.getFileName())
                .setChunks(fileChunkMetaVOS);
    }

    private void uploadChunks(MultipartFile file, MetaFile metaFile) throws IOException, ExecutionException, InterruptedException {
        List<FileChunkMeta> chunks = metaFile.getFileChunkMetas();
        //chunks = chunks.stream().filter(fileChunkMeta -> !fileChunkMeta.getCompleted()).collect(Collectors.toList());
        if (chunks.size() == 0) {
            return;
        }
        chunks = chunks.stream().sorted(Comparator.comparing(FileChunkMeta::getChunkNo)).collect(Collectors.toList());
        InputStream inputStream = file.getInputStream();
        CompletableFuture[] tasks = new CompletableFuture[chunks.size()];
        byte[] buffer = new byte[0];
        int preChunkNo = -1;
        for (int i = 0; i < chunks.size(); i++) {
            FileChunkMeta chunk = chunks.get(i);
            if (chunk.getChunkNo() != (preChunkNo)) {
                buffer = new byte[chunk.getChunkSize()];
                preChunkNo = chunk.getChunkNo();
                inputStream.read(buffer);
            }
            FileChunkBO fileChunkBO = new FileChunkBO();
            byte[] finalBuffer = buffer;
            tasks[i] = CompletableFuture.runAsync(() -> {
                if (chunk.getCompleted()) {
                    return;
                }
                String md5 = Md5Util.getMd5(finalBuffer);
                fileChunkBO.setBucketName(chunk.getBucketName())
                        .setChunkNo(chunk.getChunkNo())
                        .setChunksize(chunk.getChunkSize())
                        .setExtension(chunk.getExtension())
                        .setFilename(chunk.getFilename())
                        .setBytes(finalBuffer);
                String schema = chunk.getSchema();
                String url = chunkAddressStrategy.get(chunk);
                Object response = restTemplate.postForObject(url + "/file/write",
                        fileChunkBO,
                        Object.class);
                if (response == null) {
                    throw new RuntimeException("第" + fileChunkBO.getChunkNo() + "分片上传失败");
                }
                CommonResponse<String> md5commonResponse = objectMapper.convertValue(response, new TypeReference<CommonResponse<String>>() {
                });
                if (!md5commonResponse.getData().equals(md5)) {
                    throw new RuntimeException("文件不完整");
                }
                CompleteChunkFileBO completeChunkFileBO = new CompleteChunkFileBO()
                        .setChunkNo(chunk.getChunkNo())
                        .setAddress(chunk.getAddress())
                        .setFilename(chunk.getFilename())
                        .setSchema(chunk.getSchema())
                        .setMd5(md5);
                Object res = restTemplate.postForObject(clientConfig.getMetaServerAddress() + "/meta/chunk/complete",
                        completeChunkFileBO,
                        Object.class);
                if (res == null) {
                    throw new RuntimeException("修改metafile失败");
                }
            });

        }
        CompletableFuture<Void> allof = CompletableFuture.allOf(tasks);
        CompletableFuture<?> anyException = new CompletableFuture<>();
        Arrays.stream(tasks).forEach(t -> {
            t.exceptionally(throwable -> {
                anyException.completeExceptionally((Throwable) throwable);
                return null;
            });
        });
        CompletableFuture.anyOf(anyException, allof).get();
    }
}
